package com.swak.vertx.transport.vertx.endpoint;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.App;
import com.swak.annotation.Server;
import com.swak.reactivex.context.EndPoints.EndPoint;
import com.swak.spi.SPI;
import com.swak.utils.Ints;
import com.swak.utils.Lists;
import com.swak.utils.Maps;
import com.swak.utils.Sets;
import com.swak.vertx.config.AbstractBean;
import com.swak.vertx.config.AbstractConfig;
import com.swak.vertx.config.ImBean;
import com.swak.vertx.config.ImConfig;
import com.swak.vertx.config.RouterBean;
import com.swak.vertx.config.RouterConfig;
import com.swak.vertx.config.VertxConfig;
import com.swak.vertx.config.VertxConfigs;
import com.swak.vertx.protocol.http.ErrorHandler;
import com.swak.vertx.protocol.im.ImRouter;
import com.swak.vertx.protocol.im.ImRouter.ImRoute;
import com.swak.vertx.transport.ServerVerticle;
import com.swak.vertx.transport.vertx.HttpServerVerticle;
import com.swak.vertx.transport.vertx.ImServerVerticle;
import com.swak.vertx.transport.vertx.UnifyServerVerticle;

import io.vertx.core.CompositeFuture;
import io.vertx.core.DeploymentOptions;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.buffer.impl.BufferImpl;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.net.JksOptions;
import io.vertx.core.net.PemKeyCertOptions;
import io.vertx.core.net.PfxOptions;
import io.vertx.ext.web.Route;
import io.vertx.ext.web.Router;

/**
 * HttpEndPoint.
 * 
 * @author 618lf
 */
@SPI(value = "HttpEndPoint")
public class HttpEndPoint implements FluxEndPoint {

	Logger logger = LoggerFactory.getLogger(ServerVerticle.class);
	private VertxConfig config;
	private Vertx vertx;

	@Override
	public List<Future<EndPoint>> start(Vertx vertx, VertxConfig config) {
		this.vertx = vertx;
		this.config = config;

		List<Future<EndPoint>> futures = Lists.newArrayList();

		// 默认端口,数量转换
		this.mergeDefaults(config.getPort(), config.getWebSocketPort(), config.getEventLoopPoolSize());

		// 启动统一服务组件
		futures.addAll(this.startUnify());

		// 启动WebSocket组件
		futures.addAll(this.startWebSocket());

		// 启动Http组件
		futures.addAll(this.startHttp());

		return futures;
	}

	// ================================================================================
	// Unify
	// ================================================================================
	private List<Future<EndPoint>> startUnify() {
		List<Future<EndPoint>> futures = Lists.newArrayList();

		Map<Integer, BeansConfig> routers = this.unifys;
		Map<Integer, BeansConfig> httpRouters = this.routers;
		Map<Integer, BeansConfig> imRouters = this.webSockets;

		// 发布成多个统一服务
		routers.keySet().forEach(port -> {
			BeansConfig beansConfig = routers.get(port);
			futures.add(this.startUnify(port, beansConfig.getInstances(), httpRouters.get(port).getBeans(),
					imRouters.get(port).getBeans()));
		});
		return futures;
	}

	@SuppressWarnings("rawtypes")
	private Future<EndPoint> startUnify(int deployPort, int intstances, List<RouterBean> httpRouters,
			List<ImBean> imRouters) {

		// 发布的 Host、Port、intstances
		String deployHost = config.getHost();

		// 处理器
		Router httpRouter = this.getRouter(vertx, httpRouters);
		ImRouter imRouter = this.getImRouter(vertx, imRouters);

		// 服务器配置
		HttpServerOptions httpServerOptions = this.httpServerOptions(config);

		// 启动监听服务
		List<Future> futures = Lists.newArrayList();

		// 以EventLoop 的方式发布
		DeploymentOptions options = new DeploymentOptions().setWorker(false).setInstances(intstances);
		Future<String> stFuture = Future.future(s -> vertx.deployVerticle(
				() -> new UnifyServerVerticle(httpRouter, imRouter, httpServerOptions, deployHost, deployPort), options,
				s));
		futures.add(stFuture);

		// 合并成一个结果
		return CompositeFuture.all(futures).map(res -> {
			return new EndPoint().setScheme(Server.Unify).setPort(deployPort).setParallel(intstances);
		});
	}

	// ================================================================================
	// Http
	// ================================================================================
	private List<Future<EndPoint>> startHttp() {
		List<Future<EndPoint>> futures = Lists.newArrayList();
		Map<Integer, BeansConfig> routers = this.routers;

		// 发布成多个Http服务
		routers.keySet().forEach(port -> {
			if (!this.unifys.containsKey(port)) {
				BeansConfig beanConfig = routers.get(port);
				futures.add(this.startHttp(port, beanConfig.getInstances(), beanConfig.getBeans()));
			}
		});

		return futures;
	}

	@SuppressWarnings("rawtypes")
	private Future<EndPoint> startHttp(int deployPort, int intstances, List<RouterBean> routers) {

		// 发布的 Host、Port
		String deployHost = config.getHost();

		// 获得路由 -- Router 是线程安全的所以多个Verticle实例可以公用
		Router router = this.getRouter(vertx, routers);

		// 服务器配置
		HttpServerOptions httpServerOptions = this.httpServerOptions(config);

		// 启动监听服务
		List<Future> futures = Lists.newArrayList();

		// 以EventLoop 的方式发布
		DeploymentOptions options = new DeploymentOptions().setWorker(false).setInstances(intstances);
		Future<String> stFuture = Future.future(s -> vertx.deployVerticle(
				() -> new HttpServerVerticle(router, httpServerOptions, deployHost, deployPort), options, s));
		futures.add(stFuture);

		// 合并成一个结果
		return CompositeFuture.all(futures).map(res -> {
			return new EndPoint().setScheme(httpServerOptions.isSsl() ? Server.Https : Server.Http).setPort(deployPort)
					.setParallel(intstances);
		});
	}

	// ================================================================================
	// WebSocket
	// ================================================================================
	private List<Future<EndPoint>> startWebSocket() {
		List<Future<EndPoint>> futures = Lists.newArrayList();
		Map<Integer, BeansConfig> routers = this.webSockets;

		// 发布成多个Http服务
		routers.keySet().forEach(port -> {
			if (!this.unifys.containsKey(port)) {
				BeansConfig beansConfig = routers.get(port);
				futures.add(this.startWebSocket(port, beansConfig.getInstances(), beansConfig.getBeans()));
			}
		});

		return futures;
	}

	@SuppressWarnings("rawtypes")
	private Future<EndPoint> startWebSocket(int deployPort, int intstances, List<ImBean> routers) {

		// 发布的 Host、Port、intstances
		String deployHost = config.getHost();

		// 处理器
		ImRouter imRouter = this.getImRouter(vertx, routers);

		// 服务器配置
		HttpServerOptions httpServerOptions = this.wsServerOptions(config);

		// 启动监听服务
		List<Future> futures = Lists.newArrayList();

		// 以EventLoop 的方式发布
		DeploymentOptions options = new DeploymentOptions().setWorker(false).setInstances(intstances);
		Future<String> stFuture = Future.future(s -> vertx.deployVerticle(
				() -> new ImServerVerticle(imRouter, httpServerOptions, deployHost, deployPort), options, s));
		futures.add(stFuture);

		// 合并成一个结果
		return CompositeFuture.all(futures).map(res -> {
			return new EndPoint().setScheme(Server.Im).setPort(deployPort).setParallel(intstances);
		});
	}

	// ================================================================================
	// MergeConfigs
	// ================================================================================
	private Map<Integer, BeansConfig> unifys = Maps.newOrderMap();
	private final Map<Integer, BeansConfig> routers = Maps.newOrderMap();
	private final Map<Integer, BeansConfig> webSockets = Maps.newOrderMap();

	/**
	 * 合并默认的端口
	 * 
	 * @param httpPort  http 默认端口
	 * @param imPort    im 默认端口
	 * @param instances 默认实例
	 * @return 合并之后的配置项
	 */
	private void mergeDefaults(int httpPort, int imPort, int instances) {

		// 1. create BeansConfig
		Set<RouterBean> routerBeans = this.getRouterBeans();
		Set<ImBean> imBeans = this.getImBeans();
		for (RouterBean bean : routerBeans) {
			BeansConfig beanConfig = routers.computeIfAbsent(bean.getPort(), (port) -> {
				return new BeansConfig();
			});
			beanConfig.add(bean);
		}
		for (ImBean bean : imBeans) {
			BeansConfig beanConfig = webSockets.computeIfAbsent(bean.getPort(), (port) -> {
				return new BeansConfig();
			});
			beanConfig.add(bean);
		}

		// 2. http 端口处理 合并所有的默认配置项
		BeansConfig defaultConfig = new BeansConfig();
		Iterator<Integer> ports = routers.keySet().iterator();
		while (ports.hasNext()) {
			Integer port = ports.next();
			routers.get(port).merge(instances);
			if (port <= 0) {
				defaultConfig.merge(routers.remove(port));
			}
		}
		if (routers.containsKey(httpPort)) {
			defaultConfig.merge(routers.remove(httpPort));
		}
		if (!defaultConfig.getBeans().isEmpty()) {
			routers.putIfAbsent(httpPort, defaultConfig);
		}

		// 3. Im 端口处理 合并所有的默认配置项
		defaultConfig = new BeansConfig();
		ports = webSockets.keySet().iterator();
		while (ports.hasNext()) {
			Integer port = ports.next();
			webSockets.get(port).merge(instances);
			if (port <= 0) {
				defaultConfig.merge(webSockets.remove(port));
			}
		}
		if (webSockets.containsKey(imPort)) {
			defaultConfig.merge(webSockets.remove(imPort));
		}
		if (!defaultConfig.getBeans().isEmpty()) {
			webSockets.putIfAbsent(imPort, defaultConfig);
		}

		// 4. 统一处理器
		routers.keySet().forEach(port -> {
			if (webSockets.containsKey(port)) {
				BeansConfig unifyConfig = unifys.computeIfAbsent(port, (key) -> {
					return new BeansConfig();
				});
				unifyConfig.instances = Ints.max(routers.get(port).instances, webSockets.get(port).instances);
			}
		});
	}

	// ================================================================================
	// Configs
	// ================================================================================
	private Set<RouterBean> getRouterBeans() {
		Set<RouterBean> services = Sets.newOrderSet();
		Set<AbstractConfig> configs = VertxConfigs.me().getConfigs();
		for (AbstractConfig config : configs) {
			if (config instanceof RouterBean) {
				services.add((RouterBean) config);
			}
		}
		return services;
	}

	private Set<ImBean> getImBeans() {
		Set<ImBean> services = Sets.newOrderSet();
		Set<AbstractConfig> configs = VertxConfigs.me().getConfigs();
		for (AbstractConfig config : configs) {
			if (config instanceof ImBean) {
				services.add((ImBean) config);
			}
		}
		return services;
	}

	private Set<ImConfig> getImConfigs() {
		Set<ImConfig> services = Sets.newOrderSet();
		Set<AbstractConfig> configs = VertxConfigs.me().getConfigs();
		for (AbstractConfig config : configs) {
			if (config instanceof ImConfig) {
				services.add((ImConfig) config);
			}
		}
		return services;
	}

	private Set<RouterConfig> getHttpConfigs() {
		Set<RouterConfig> services = Sets.newOrderSet();
		Set<AbstractConfig> configs = VertxConfigs.me().getConfigs();
		for (AbstractConfig config : configs) {
			if (config instanceof RouterConfig) {
				services.add((RouterConfig) config);
			}
		}
		return services;
	}

	private ImRouter getImRouter(Vertx vertx, List<ImBean> routers) {

		// Im 的 Router
		ImRouter imRouter = new ImRouter();

		// 路由基本配置
		Set<ImConfig> configs = this.getImConfigs();
		for (ImConfig config : configs) {
			config.apply(vertx, imRouter);
		}

		// 单个路由定义
		for (ImBean rb : routers) {
			rb.mounton(imRouter);
			rb.export();
		}

		// 打印路由信息
		if (logger.isDebugEnabled()) {
			List<ImRoute> routes = imRouter.getRoutes();
			for (ImRoute route : routes) {
				logger.debug("{}", route.getOps() != null ? route.getOps().toString() : "All");
			}
		}
		return imRouter;
	}

	private Router getRouter(Vertx vertx, List<RouterBean> routers) {

		// 初始化
		Router router = Router.router(vertx);
		router.errorHandler(404, new ErrorHandler(404));
		router.errorHandler(500, new ErrorHandler(500));

		// 路由基本配置
		Set<RouterConfig> configs = this.getHttpConfigs();
		for (RouterConfig config : configs) {
			config.apply(vertx, router);
		}

		// 单个路由定义
		for (RouterBean rb : routers) {
			rb.mounton(router);
			rb.export();
		}

		// 打印路由信息
		if (logger.isDebugEnabled()) {
			List<Route> routes = router.getRoutes();
			for (Route route : routes) {
				logger.debug("{}\t{}", route.methods() != null ? route.methods().toString() : "All",
						route.getPath() != null ? route.getPath() : "All");
			}
		}
		return router;
	}

	/**
	 * 服务的配置
	 * 
	 * @param properties
	 * @return
	 */
	private HttpServerOptions httpServerOptions(VertxConfig config) {

		HttpServerOptions options = new HttpServerOptions();

		if (config.isCompressionSupported()) {
			options.setCompressionSupported(config.isCompressionSupported());
		}
		if (config.getCompressionLevel() != -1) {
			options.setCompressionLevel(config.getCompressionLevel());
		}
		if (config.getKeyPaths() != null && config.getCertPaths() != null && config.getKeyPaths().size() > 0
				&& config.getCertPaths().size() == config.getKeyPaths().size()) {
			PemKeyCertOptions pemKeys = null;
			JksOptions jksKeys = null;
			PfxOptions pfxOptions = null;
			for (int i = 0; i < config.getKeyPaths().size(); i++) {
				String keyPath = config.getKeyPaths().get(i);
				String certPath = config.getCertPaths().get(i);

				if (keyPath.endsWith(".pem")) {
					Buffer keyValue = BufferImpl.buffer(App.bytes(keyPath));
					Buffer certValue = BufferImpl.buffer(App.bytes(certPath));
					if (pemKeys == null) {
						pemKeys = new PemKeyCertOptions();
					}
					pemKeys.addKeyValue(keyValue).addCertValue(certValue);
				}

				else if (keyPath.endsWith(".pfx")) {
					if (pfxOptions == null) {
						pfxOptions = new PfxOptions();
					}
					Buffer keyValue = BufferImpl.buffer(App.bytes(keyPath));
					jksKeys.setValue(keyValue).setPassword(certPath);
				}

				else if (keyPath.endsWith(".jks")) {
					if (jksKeys == null) {
						jksKeys = new JksOptions();
					}
					Buffer keyValue = BufferImpl.buffer(App.bytes(keyPath));
					jksKeys.setValue(keyValue).setPassword(certPath);
				}
			}
			if (pemKeys != null) {
				options.setPemKeyCertOptions(pemKeys);
			}
			if (pfxOptions != null) {
				options.setPfxKeyCertOptions(pfxOptions);
			}
			if (jksKeys != null) {
				options.setKeyStoreOptions(jksKeys);
			}
		}
		options.setSsl(config.isUseSsl());
		options.setUseAlpn(config.isUseAlpn());
		options.setClientAuth(config.getClientAuth());
		options.setMaxHeaderSize(config.getMaxHeaderSize());
		options.setMaxFormAttributeSize(config.getMaxFormAttributeSize());
		return options;
	}
	
	/**
	 * 服务的配置
	 * 
	 * @param properties
	 * @return
	 */
	private HttpServerOptions wsServerOptions(VertxConfig config) {

		HttpServerOptions options = new HttpServerOptions();

		if (config.isCompressionSupported()) {
			options.setCompressionSupported(config.isCompressionSupported());
		}
		if (config.getCompressionLevel() != -1) {
			options.setCompressionLevel(config.getCompressionLevel());
		}
		if (config.getKeyPaths() != null && config.getCertPaths() != null && config.getKeyPaths().size() > 0
				&& config.getCertPaths().size() == config.getKeyPaths().size()) {
			PemKeyCertOptions pemKeys = null;
			JksOptions jksKeys = null;
			PfxOptions pfxOptions = null;
			for (int i = 0; i < config.getKeyPaths().size(); i++) {
				String keyPath = config.getKeyPaths().get(i);
				String certPath = config.getCertPaths().get(i);

				if (keyPath.endsWith(".pem")) {
					Buffer keyValue = BufferImpl.buffer(App.bytes(keyPath));
					Buffer certValue = BufferImpl.buffer(App.bytes(certPath));
					if (pemKeys == null) {
						pemKeys = new PemKeyCertOptions();
					}
					pemKeys.addKeyValue(keyValue).addCertValue(certValue);
				}

				else if (keyPath.endsWith(".pfx")) {
					if (pfxOptions == null) {
						pfxOptions = new PfxOptions();
					}
					Buffer keyValue = BufferImpl.buffer(App.bytes(keyPath));
					jksKeys.setValue(keyValue).setPassword(certPath);
				}

				else if (keyPath.endsWith(".jks")) {
					if (jksKeys == null) {
						jksKeys = new JksOptions();
					}
					Buffer keyValue = BufferImpl.buffer(App.bytes(keyPath));
					jksKeys.setValue(keyValue).setPassword(certPath);
				}
			}
			if (pemKeys != null) {
				options.setPemKeyCertOptions(pemKeys);
			}
			if (pfxOptions != null) {
				options.setPfxKeyCertOptions(pfxOptions);
			}
			if (jksKeys != null) {
				options.setKeyStoreOptions(jksKeys);
			}
		}
		options.setSsl(config.isUseSsl());
		options.setUseAlpn(config.isUseAlpn());
		options.setClientAuth(config.getClientAuth());
		options.setMaxHeaderSize(config.getMaxHeaderSize());
		options.setMaxFormAttributeSize(config.getMaxFormAttributeSize());
		options.setIdleTimeout(config.getWebSocketHeartbeatMs());
		options.setIdleTimeoutUnit(TimeUnit.MILLISECONDS);
		return options;
	}

	public static class BeansConfig {
		private int instances = -1;
		private List<AbstractBean> beans = new LinkedList<>();

		public int getInstances() {
			return instances;
		}

		public void setInstances(int instances) {
			this.instances = instances;
		}

		@SuppressWarnings("unchecked")
		public <T> List<T> getBeans() {
			return (List<T>) beans;
		}

		public void setBeans(List<AbstractBean> beans) {
			this.beans = beans;
		}

		public BeansConfig add(AbstractBean bean) {
			this.beans.add(bean);
			this.instances = Ints.max(this.instances, bean.getInstances());
			return this;
		}

		public BeansConfig merge(BeansConfig config) {
			this.beans.addAll(config.beans);
			this.instances = Ints.max(this.instances, config.instances);
			return this;
		}

		public BeansConfig merge(int instances) {
			this.instances = this.instances <= 0 ? instances : this.instances;
			return this;
		}
	}
}
